﻿using Google.Protobuf;
using Grpc.Core;
using MX.Core.Log;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace MX.Proto
{
    public class Publisher
    {
        static Dictionary<string, Subscriber<PublishMessage>> subscribeList = new Dictionary<string, Subscriber<PublishMessage>>();
        const int maxErrCount = 5;
        Func<string, string, Notify> getNotifyFun;
        Task worker;
        Action<string> ackAction;


        public Publisher(Func<string, string, Notify> getOneNotify, Action<string> ack)
        {
            ackAction = ack;
            getNotifyFun = getOneNotify;
            worker = new Task(() =>
            {
                Run();
            });
        } 

        private void Run()
        {
            while (true)
            { 
                try
                {
                    foreach (var item in subscribeList.ToArray())
                    {
                        var notify = getNotifyFun(item.Value.AccessId, item.Value.Name);
                        if (notify != null)
                        {
                            item.Value.StreamList.ToArray().AsParallel().ForAll(async (stream) =>
                            {
                                if (!stream.PreSuccess && DateTime.Now > DateTime.Now.AddSeconds(Math.Pow(stream.ErrorCount, 2))) return;
                                try
                                {
                                    stream.PreActiveTime = DateTime.Now;
                                    var message = new PublishMessage()
                                    {
                                        Message = ByteString.FromBase64(notify.Message),
                                        Name = notify.SubscribeName,
                                        PublishId = notify.NotifyId
                                    };
                                    Logger.GetLogger(nameof(Publisher)).Debug($"写入消息到客户端{stream.Peer}:\n{message}");
                                    await stream.Writer.WriteAsync(message);
                                }
                                catch (Exception ex)
                                {
                                    stream.ErrorCount++;
                                    stream.PreSuccess = false;
                                    Logger.GetLogger(nameof(Publisher)).Error(ex, $"客户端{stream.Peer}:已错误{stream.ErrorCount}");
                                }
                            });
                            var needRemove = item.Value.StreamList.Where(s => s.ErrorCount > maxErrCount).ToArray();
                            if (needRemove.Count() >= item.Value.StreamList.Count)
                            {
                                subscribeList.Remove(item.Key);
                            }
                            else
                            {
                                foreach (var stream in needRemove)
                                {
                                    item.Value.StreamList.Remove(stream);
                                }
                            }
                        }
                    }
                }
                catch (Exception ex)
                {
                    Logger.GetLogger(nameof(Publisher)).Error(ex, "");
                }
                finally
                {
                } 
            }
        }

        public void Ack(string name, string accessId, string notifyId)
        {
            try
            {
                ackAction(notifyId);
            }
            catch (Exception ex)
            {
                Logger.GetLogger(nameof(Publisher)).Error(ex, $"ACK失败:{notifyId}");
            }
        }

        public void Subscribe(string name, string accessId, string peer, IServerStreamWriter<PublishMessage> writer)
        {
            string key = $"SDK_GRPC_Subscribe_Notify_{name}_{accessId}";
            if (!subscribeList.ContainsKey(key))
            {
                subscribeList[key] = new Subscriber<PublishMessage>()
                {
                    AccessId = accessId,
                    Name = name,
                    StreamList = new List<StreamInfo<PublishMessage>>()
                };
            }
            subscribeList[key].StreamList.Add(new StreamInfo<PublishMessage>()
            {
                CreateTime = DateTime.Now,
                ErrorCount = 0,
                Peer = peer,
                PreActiveTime = DateTime.Now,
                Writer = writer,
                PreSuccess = true
            });
            if (worker.Status == TaskStatus.Created) { worker.Start(); }
        }
    }
}
